package top.zhaoyuliu.logbook.client.queue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**  
 * Copyright © 2021 Arain. All rights reserved.
 *
 * @Title: LbQueue.java
 * @Prject: logbook-client
 * @Package: top.zhaoyuliu.logbook.client.queue
 * @Description: <功能详细描述>
 * @author: Arain  
 * @date: 2021年1月23日 上午10:42:19
 * @version: V1.0  
 */

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.tio.client.ClientGroupContext;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import top.zhaoyuliu.logbook.client.builder.LbBuilder;
import top.zhaoyuliu.logbook.client.builder.ParameterLbBuilder;
import top.zhaoyuliu.logbook.client.builder.TaskInfo;
import top.zhaoyuliu.logbook.client.compent.AppenderCompent;
import top.zhaoyuliu.logbook.client.connect.packet.PacketBuilder;
import top.zhaoyuliu.logbook.client.connect.packet.PacketType;
import top.zhaoyuliu.logbook.client.constant.Const;
import top.zhaoyuliu.logbook.client.link.LinkObject;
import top.zhaoyuliu.logbook.client.thread.ThreadPool;
import top.zhaoyuliu.logbook.client.utils.IPUtils;

@Slf4j
public class LbQueue {

	public static int pushSize = 50;

	private static BlockingQueue<LbBuilder> logQueue = new LinkedBlockingQueue<>();

	private static BlockingQueue<LbBuilder> paraQueue = new LinkedBlockingQueue<>();

	private static BlockingQueue<LbBuilder> trackQueue = new LinkedBlockingQueue<>();

	public static void pushLogQueue(LbBuilder builder) {
		try {
			builder.setIp(IPUtils.getIP());
			builder.setPort(AppenderCompent.getPort());
			logQueue.add(builder);
			sendData(logQueue, PacketType.LOG);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void pushtrackQueue(LbBuilder builder) {
		try {
			builder.setIp(IPUtils.getIP());
			builder.setPort(AppenderCompent.getPort());
			trackQueue.add(builder);
			sendData(trackQueue, PacketType.TRACK);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void pushParaQueue(ParameterLbBuilder builder) {
		try {
			builder.setIp(IPUtils.getIP());
			builder.setPort(AppenderCompent.getPort());
			paraQueue.add(builder);
			sendDataSet(paraQueue, PacketType.PARA);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void sendData(BlockingQueue<LbBuilder> data, byte type) {
		if (data.size() >= pushSize) {
			try {
				List<LbBuilder> list = new ArrayList<>();
				int drainTo = data.drainTo(list);
				if (drainTo <= 0) {
					LbBuilder log = data.take();
					list.add(log);
				}
				checkDataAndSend(list, type);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private static void sendDataSet(BlockingQueue<LbBuilder> data, byte type) {
		if (data.size() >= pushSize) {
			try {
				Set<LbBuilder> list = new HashSet<>();
				int drainTo = data.drainTo(list);
				if (drainTo <= 0) {
					LbBuilder log = data.take();
					list.add(log);
				}
				checkDataAndSend(list, type);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static void pushParaQueue(String json) {
		ThreadPool.INSTANCE.executor().execute(new Runnable() {
			@Override
			public void run() {
				try {
					ParameterLbBuilder builder = JSONUtil.toBean(json, ParameterLbBuilder.class);
					builder.setIp(IPUtils.getIP());
					builder.setPort(AppenderCompent.getPort());
					paraQueue.add(builder);
					sendDataSet(paraQueue, PacketType.PARA);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

		});

	}

	public static void pushtrackQueue(String jsonStr) {
		ThreadPool.INSTANCE.executor().execute(new Runnable() {
			@Override
			public void run() {
				try {
					LinkObject builder = JSONUtil.toBean(jsonStr, LinkObject.class);
					builder.setIp(IPUtils.getIP());
					builder.setPort(AppenderCompent.getPort());
					trackQueue.add(builder);
					sendData(trackQueue, PacketType.TRACK);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
	}

	public static void pushLogQueue(String json) {
		ThreadPool.INSTANCE.executor().execute(new Runnable() {
			@Override
			public void run() {
				try {
					LbBuilder builder = JSONUtil.toBean(json, LbBuilder.class);
					builder.setIp(IPUtils.getIP());
					builder.setPort(AppenderCompent.getPort());
					logQueue.add(builder);
					sendData(logQueue, PacketType.LOG);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

		});
	}

	private static void checkDataAndSend(Collection<LbBuilder> data, byte type) {
		if (Const.CLUSTER) {
			ClientGroupContext groupContext = AppenderCompent.getGroup();
			if (null != groupContext) {
				ChannelContext context = AppenderCompent.choose();
				Tio.send(context, new PacketBuilder().setBody(data).setType(type).build());
			}
		} else {
			Tio.send(AppenderCompent.getClientChannelContext(),
					new PacketBuilder().setBody(data).setType(type).build());
		}
	}

	public static void sendToCluster(TaskInfo info, ChannelContext channelContext) {
		if (ObjectUtil.isNotNull(info)) {
			try {
				Byte type = AppenderCompent.typeCache.get(info.getKey());
				Collection<LbBuilder> collection = AppenderCompent.weakCache.get(info.getKey());
				Integer count = info.getCount();
				List<Integer> list = AppenderCompent.taskCountCache.get(info.getKey());
				if (CollectionUtil.isEmpty(list)) {
					list = new ArrayList<>();
				}
				list.add(count);
				AppenderCompent.taskCountCache.put(info.getKey(), list);
				Thread.sleep(Const.TASKTIMEOUT);
				List<Integer> counts = AppenderCompent.taskCountCache.get(info.getKey());
				if (CollectionUtil.isEmpty(counts)) {
					Tio.send(channelContext, new PacketBuilder().setBody(collection).setType(type).build());
				}
				CollectionUtil.sort(counts, new Comparator<Integer>() {
					@Override
					public int compare(Integer o1, Integer o2) {
						if (o1.intValue() > o2.intValue()) {
							return 1;
						} else if (o1.intValue() < o2.intValue()) {
							return -1;
						}
						return 0;
					}
				});
				Integer integer = counts.get(0);
				if (info.getCount().intValue() == integer.intValue()) {
					Tio.send(channelContext, new PacketBuilder().setBody(collection).setType(type).build());
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} else {
			log.warn("日志丢弃");
		}

	}
}
